ScaDaMaLe Course site and book

This notebook is from databricks document: - https://docs.databricks.com/_static/notebooks/package-cells.html

As you know, we need to eventually package and deploy our models, say using sbt.

However it is nice to be in a notebook environment to prototype and build intuition and create better pipelines.

Using package cells we can be in the best of both worlds to an extent.

Package Cells

Package cells are special cells that get compiled when executed. These cells have no visibility with respect to the rest of the notebook. You may think of them as separate scala files.

This means that only class and object definitions may go inside this cell. You may not have any variable or function definitions lying around by itself. The following cell will not work.

If you wish to use custom classes and/or objects defined within notebooks reliably in Spark, and across notebook sessions, you must use package cells to define those classes.

Unless you use package cells to define classes, you may also come across obscure bugs as follows:

// We define a class
case class TestKey(id: Long, str: String)

defined class TestKey

// we use that class as a key in the group by
val rdd = sc.parallelize(Array((TestKey(1L, "abd"), "dss"), (TestKey(2L, "ggs"), "dse"), (TestKey(1L, "abd"), "qrf")))

rdd.groupByKey().collect

rdd: org.apache.spark.rdd.RDD[(TestKey, String)] = ParallelCollectionRDD[0] at parallelize at command-2971213210276613:2
res0: Array[(TestKey, Iterable[String])] = Array((TestKey(1,abd),CompactBuffer(dss)), (TestKey(1,abd),CompactBuffer(qrf)), (TestKey(2,ggs),CompactBuffer(dse)))

What went wrong above? Even though we have two elements for the key TestKey(1L, "abd"), they behaved as two different keys resulting in:

Array[(TestKey, Iterable[String])] = Array(
  (TestKey(2,ggs),CompactBuffer(dse)), 
  (TestKey(1,abd),CompactBuffer(dss)), 
  (TestKey(1,abd),CompactBuffer(qrf)))

Once we define our case class within a package cell, we will not face this issue.

package com.databricks.example

case class TestKey(id: Long, str: String)

Warning: classes defined within packages cannot be redefined without a cluster restart.
Compilation successful.

import com.databricks.example

val rdd = sc.parallelize(Array(
  (example.TestKey(1L, "abd"), "dss"), (example.TestKey(2L, "ggs"), "dse"), (example.TestKey(1L, "abd"), "qrf")))

rdd.groupByKey().collect

import com.databricks.example
rdd: org.apache.spark.rdd.RDD[(com.databricks.example.TestKey, String)] = ParallelCollectionRDD[2] at parallelize at command-2971213210276616:3
res2: Array[(com.databricks.example.TestKey, Iterable[String])] = Array((TestKey(1,abd),CompactBuffer(dss, qrf)), (TestKey(2,ggs),CompactBuffer(dse)))

As you can see above, the group by worked above, grouping two elements (dss, qrf) under TestKey(1,abd).

These cells behave as individual source files, therefore only classes and objects can be defined inside these cells.

package x.y.z

val aNumber = 5 // won't work

def functionThatWillNotWork(a: Int): Int = a + 1

The following cell is the way to go.

package x.y.z

object Utils {
  val aNumber = 5 // works!
  def functionThatWillWork(a: Int): Int = a + 1
}

Warning: classes defined within packages cannot be redefined without a cluster restart.
Compilation successful.

import x.y.z.Utils

Utils.functionThatWillWork(Utils.aNumber)

import x.y.z.Utils
res5: Int = 6

Why did we get the warning: classes defined within packages cannot be redefined without a cluster restart?

Classes that get compiled with the package cells get dynamically injected into Spark's classloader. Currently it's not possible to remove classes from Spark's classloader. Any classes that you define and compile will have precedence in the classloader, therefore once you recompile, it will not be visible to your application.

Well that kind of beats the purpose of iterative notebook development if I have to restart the cluster, right? In that case, you may just rename the package to x.y.z2 during development/fast iteration and fix it once everything works.

One thing to remember with package cells is that it has no visiblity regarding the notebook environment.

• The SparkContext will not be defined as sc.
• The SQLContext will not be defined as sqlContext.
• Did you import a package in a separate cell? Those imports will not be available in the package cell and have to be remade.
• Variables imported through %run cells will not be available.

It is really a standalone file that just looks like a cell in a notebook. This means that any function that uses anything that was defined in a separate cell, needs to take that variable as a parameter or the class needs to take it inside the constructor.

package x.y.zpackage

import org.apache.spark.SparkContext

case class IntArray(values: Array[Int])

class MyClass(sc: SparkContext) {
  def sparkSum(array: IntArray): Int = {
    sc.parallelize(array.values).reduce(_ + _)
  }
}

object MyClass {
  def sparkSum(sc: SparkContext, array: IntArray): Int = {
    sc.parallelize(array.values).reduce(_ + _)
  }
}

Warning: classes defined within packages cannot be redefined without a cluster restart.
Compilation successful.

import x.y.zpackage._

val array = IntArray(Array(1, 2, 3, 4, 5))

val myClass = new MyClass(sc)
myClass.sparkSum(array)

import x.y.zpackage._
array: x.y.zpackage.IntArray = IntArray([I@44a91426)
myClass: x.y.zpackage.MyClass = x.y.zpackage.MyClass@5d3b142
res7: Int = 15

MyClass.sparkSum(sc, array)

res8: Int = 15

Build Packages Locally

Although package cells are quite handy for quick prototyping in notebook environemnts, it is better to develop packages locally on your laptop and then upload the packaged or assembled jar file into databricks to use the classes and methods developed in the package.

ScaDaMaLe Course site and book

Introduction to Simulation

Core ideas in Monte Carlo simulation

• modular arithmetic gives pseudo-random streams that are indistiguishable from 'true' Uniformly distributed samples in integers from \({0,1,2,...,m}\)
• by diving the integer streams from above by \(m\) we get samples from \({0/m,1/m,...,(m-1)/m}\) and "pretend" this to be samples from the Uniform(0,1) RV
• we can use inverse distribution function of von Neumann's rejection sampler to convert samples from Uniform(0,1) RV to the following:
  • any other random variable
  • vector of random variables that could be dependent
  • or more generally other random structures:
    • random graphs and networks
    • random walks or (sensible perturbations of live traffic data on open street maps for hypothesis tests)
    • models of interacting paticle systems in ecology / chemcal physics, etc...

• https://en.wikipedia.org/wiki/Inversetransformsampling
• https://en.wikipedia.org/wiki/Rejection_sampling - will revisit below for Expoential RV

breeze.stats.distributions

Breeze also provides a fairly large number of probability distributions. These come with access to probability density function for either discrete or continuous distributions. Many distributions also have methods for giving the mean and the variance.

import breeze.stats.distributions._

val poi = new Poisson(3.0);

import breeze.stats.distributions._
poi: breeze.stats.distributions.Poisson = Poisson(3.0)

val s = poi.sample(5); // let's draw five samples - black-box

s: IndexedSeq[Int] = Vector(2, 4, 3, 4, 6)

Getting probabilities of the Poisson samples

s.map( x => poi.probabilityOf(x) ) // PMF

res0: IndexedSeq[Double] = Vector(0.22404180765538775, 0.16803135574154085, 0.22404180765538775, 0.16803135574154085, 0.05040940672246224)

val doublePoi = for(x <- poi) yield x.toDouble // meanAndVariance requires doubles, but Poisson samples over Ints

doublePoi: breeze.stats.distributions.Rand[Double] = MappedRand(Poisson(3.0),$Lambda$5909/1062631843@70cdbd4c)

breeze.stats.meanAndVariance(doublePoi.samples.take(1000));

res1: breeze.stats.meanAndVariance.MeanAndVariance = MeanAndVariance(3.0660000000000034,3.250894894894892,1000)

(poi.mean, poi.variance) // population mean and variance

res2: (Double, Double) = (3.0,3.0)

Exponential random Variable

Let's focus on getting our hands direty with a common random variable.

NOTE: Below, there is a possibility of confusion for the term rate in the family of exponential distributions. Breeze parameterizes the distribution with the mean, but refers to it as the rate.

val expo = new Exponential(0.5);

expo: breeze.stats.distributions.Exponential = Exponential(0.5)

expo.rate // what is the rate parameter

res3: Double = 0.5

A characteristic of exponential distributions is its half-life, but we can compute the probability a value falls between any two numbers.

expo.probability(0, math.log(2) * expo.rate)

res4: Double = 0.5

expo.probability(math.log(2) * expo.rate, 10000.0)

res5: Double = 0.5

expo.probability(0.0, 1.5)

res6: Double = 0.950212931632136

The above result means that approximately 95% of the draws from an exponential distribution fall between 0 and thrice the mean. We could have easily computed this with the cumulative distribution as well.

1 - math.exp(-3.0) // the CDF of the Exponential RV with rate parameter 3

res7: Double = 0.950212931632136

Drawing samples from Exponential RV

val samples = expo.sample(2).sorted; // built-in black box - we will roll our own shortly in Spark

samples: IndexedSeq[Double] = Vector(2.412493444974671, 3.6802759985818687)

expo.probability(samples(0), samples(1));

res8: Double = 0.007390811722912227

breeze.stats.meanAndVariance(expo.samples.take(10000)); // mean and variance of the sample

res9: breeze.stats.meanAndVariance.MeanAndVariance = MeanAndVariance(2.0109026469746554,4.073504140726905,10000)

(1 / expo.rate, 1 / (expo.rate * expo.rate)) // mean and variance of the population

res10: (Double, Double) = (2.0,4.0)

Pseudo Random Numbers in Spark

import spark.implicits._
import org.apache.spark.sql.functions._

import spark.implicits._
import org.apache.spark.sql.functions._

val df = spark.range(1000).toDF("Id") // just make a DF of 1000 row indices

df: org.apache.spark.sql.DataFrame = [Id: bigint]

df.show(5)

+---+
| Id|
+---+
|  0|
|  1|
|  2|
|  3|
|  4|
+---+
only showing top 5 rows

val dfRand = df.select($"Id", rand(seed=1234567) as "rand") // add a column of random numbers in (0,1)

dfRand: org.apache.spark.sql.DataFrame = [Id: bigint, rand: double]

dfRand.show(5) // these are first 5 of the 1000 samples from the Uniform(0,1) RV

+---+--------------------+
| Id|                rand|
+---+--------------------+
|  0|0.024042816995877625|
|  1|  0.4763832311243641|
|  2|  0.3392911406256911|
|  3|  0.6282132043405342|
|  4|  0.9665960987271114|
+---+--------------------+
only showing top 5 rows

val dfRand = df.select($"Id", rand(seed=1234567) as "rand") // add a column of random numbers in (0,1)
dfRand.show(5) // these are first 5 of the 1000 samples from the Uniform(0,1) RV

+---+--------------------+
| Id|                rand|
+---+--------------------+
|  0|0.024042816995877625|
|  1|  0.4763832311243641|
|  2|  0.3392911406256911|
|  3|  0.6282132043405342|
|  4|  0.9665960987271114|
+---+--------------------+
only showing top 5 rows

dfRand: org.apache.spark.sql.DataFrame = [Id: bigint, rand: double]

val dfRand = df.select($"Id", rand(seed=879664) as "rand") // add a column of random numbers in (0,1)
dfRand.show(5) // these are first 5 of the 1000 samples from the Uniform(0,1) RV

+---+-------------------+
| Id|               rand|
+---+-------------------+
|  0|  0.269224348263137|
|  1|0.20433965628039852|
|  2| 0.8481228617934538|
|  3| 0.6665103087537884|
|  4| 0.7712898780552342|
+---+-------------------+
only showing top 5 rows

dfRand: org.apache.spark.sql.DataFrame = [Id: bigint, rand: double]

Let's use the inverse CDF of the Exponential RV to transform these samples from the Uniform(0,1) RV into those from the Exponential RV.

val dfRand = df.select($"Id", rand(seed=1234567) as "rand") // add a column of random numbers in (0,1)
               .withColumn("one",lit(1.0))
               .withColumn("rate",lit(0.5))

dfRand: org.apache.spark.sql.DataFrame = [Id: bigint, rand: double ... 2 more fields]

dfRand.show(5) 

+---+--------------------+---+----+
| Id|                rand|one|rate|
+---+--------------------+---+----+
|  0|0.024042816995877625|1.0| 0.5|
|  1|  0.4763832311243641|1.0| 0.5|
|  2|  0.3392911406256911|1.0| 0.5|
|  3|  0.6282132043405342|1.0| 0.5|
|  4|  0.9665960987271114|1.0| 0.5|
+---+--------------------+---+----+
only showing top 5 rows

val dfExpRand = dfRand.withColumn("expo_sample", -($"one" / $"rate") * log($"one" - $"rand")) // samples from expo(rate=0.5)

dfExpRand: org.apache.spark.sql.DataFrame = [Id: bigint, rand: double ... 3 more fields]

dfExpRand.show(5)

+---+--------------------+---+----+--------------------+
| Id|                rand|one|rate|         expo_sample|
+---+--------------------+---+----+--------------------+
|  0|0.024042816995877625|1.0| 0.5|0.048673126808362034|
|  1|  0.4763832311243641|1.0| 0.5|  1.2939904386814756|
|  2|  0.3392911406256911|1.0| 0.5|  0.8288839819270684|
|  3|  0.6282132043405342|1.0| 0.5|  1.9788694379168241|
|  4|  0.9665960987271114|1.0| 0.5|   6.798165162486205|
+---+--------------------+---+----+--------------------+
only showing top 5 rows

display(dfExpRand)

dfExpRand.describe().show() // look sensible

+-------+-----------------+--------------------+----+----+--------------------+
|summary|               Id|                rand| one|rate|         expo_sample|
+-------+-----------------+--------------------+----+----+--------------------+
|  count|             1000|                1000|1000|1000|                1000|
|   mean|            499.5|  0.5060789944940257| 1.0| 0.5|   2.024647367046763|
| stddev|288.8194360957494| 0.28795893414434537| 0.0| 0.0|  1.9649069025265538|
|    min|                0|6.282087728304298E-4| 1.0| 0.5|0.001256812357281...|
|    max|              999|  0.9981606891187609| 1.0| 0.5|  12.596728597154938|
+-------+-----------------+--------------------+----+----+--------------------+

val expoSamplesDF = spark.range(1000000000).toDF("Id") // just make a DF of 100 row indices
               .select($"Id", rand(seed=1234567) as "rand") // add a column of random numbers in (0,1)
               .withColumn("one",lit(1.0))
               .withColumn("rate",lit(0.5))
               .withColumn("expo_sample", -($"one" / $"rate") * log($"one" - $"rand"))

expoSamplesDF: org.apache.spark.sql.DataFrame = [Id: bigint, rand: double ... 3 more fields]

expoSamplesDF.describe().show()

+-------+--------------------+--------------------+----------+----------+--------------------+
|summary|                  Id|                rand|       one|      rate|         expo_sample|
+-------+--------------------+--------------------+----------+----------+--------------------+
|  count|          1000000000|          1000000000|1000000000|1000000000|          1000000000|
|   mean|4.9999999907595944E8| 0.49999521358240573|       1.0|       0.5|  1.9999814119383708|
| stddev|2.8867513473915565E8|  0.2886816216562552|       0.0|       0.0|  2.0000011916404206|
|    min|                   0|1.584746778249268...|       1.0|       0.5|3.169493556749679...|
|    max|           999999999|  0.9999999996809935|       1.0|       0.5|  43.731619350261035|
+-------+--------------------+--------------------+----------+----------+--------------------+

Approximating Pi with Monte Carlo simulations

Uisng RDDs directly, let's estimate Pi.

//Calculate pi with Monte Carlo estimation
import scala.math.random

//make a very large unique set of 1 -> n 
val partitions = 2 
val n = math.min(100000L * partitions, Int.MaxValue).toInt 
val xs = 1 until n 

//split up n into the number of partitions we can use 
val rdd = sc.parallelize(xs, partitions).setName("'N values rdd'")

//generate a random set of points within a 2x2 square
val sample = rdd.map { i =>
  val x = random * 2 - 1
  val y = random * 2 - 1
  (x, y)
}.setName("'Random points rdd'")

//points w/in the square also w/in the center circle of r=1
val inside = sample.filter { case (x, y) => (x * x + y * y < 1) }.setName("'Random points inside circle'")
val count = inside.count()
 
//Area(circle)/Area(square) = inside/n => pi=4*inside/n                        
println("Pi is roughly " + 4.0 * count / n)

Pi is roughly 3.14156
import scala.math.random
partitions: Int = 2
n: Int = 200000
xs: scala.collection.immutable.Range = Range 1 until 200000
rdd: org.apache.spark.rdd.RDD[Int] = 'N values rdd' ParallelCollectionRDD[39] at parallelize at command-2971213210276568:10
sample: org.apache.spark.rdd.RDD[(Double, Double)] = 'Random points rdd' MapPartitionsRDD[40] at map at command-2971213210276568:13
inside: org.apache.spark.rdd.RDD[(Double, Double)] = 'Random points inside circle' MapPartitionsRDD[41] at filter at command-2971213210276568:20
count: Long = 157078

Doing it in PySpark is just as easy. This may be needed if there are pyhton libraries you want to take advantage of in Spark.

# # Estimating $\pi$
#
# This PySpark example shows you how to estimate $\pi$ in parallel
# using Monte Carlo integration.

from __future__ import print_function
import sys
from random import random
from operator import add

partitions = 2
n = 100000 * partitions

def f(_):
    x = random() * 2 - 1
    y = random() * 2 - 1
    return 1 if x ** 2 + y ** 2 < 1 else 0

# To access the associated SparkContext
count = spark.sparkContext.parallelize(range(1, n + 1), partitions).map(f).reduce(add)
print("Pi is roughly %f" % (4.0 * count / n))

Pi is roughly 3.138820

The following is from this google turotial.

Programming a Monte Carlo simulation in Scala

Monte Carlo, of course, is famous as a gambling destination. In this section, you use Scala to create a simulation that models the mathematical advantage that a casino enjoys in a game of chance. The "house edge" at a real casino varies widely from game to game; it can be over 20% in keno, for example. This tutorial creates a simple game where the house has only a one-percent advantage. Here's how the game works:

1. The player places a bet, consisting of a number of chips from a bankroll fund.

• The player rolls a 100-sided die (how cool would that be?).
• If the result of the roll is a number from 1 to 49, the player wins.
• For results 50 to 100, the player loses the bet.

You can see that this game creates a one-percent disadvantage for the player: in 51 of the 100 possible outcomes for each roll, the player loses.

Follow these steps to create and run the game:

val STARTING_FUND = 10
val STAKE = 1   // the amount of the bet
val NUMBER_OF_GAMES = 25

def rollDie: Int = {
    val r = scala.util.Random
    r.nextInt(99) + 1
}

def playGame(stake: Int): (Int) = {
    val faceValue = rollDie
    if (faceValue < 50)
        (2*stake)
    else
        (0)
}

// Function to play the game multiple times
// Returns the final fund amount
def playSession(
   startingFund: Int = STARTING_FUND,
   stake: Int = STAKE,
   numberOfGames: Int = NUMBER_OF_GAMES):
   (Int) = {

    // Initialize values
    var (currentFund, currentStake, currentGame) = (startingFund, 0, 1)

    // Keep playing until number of games is reached or funds run out
    while (currentGame <= numberOfGames && currentFund > 0) {

        // Set the current bet and deduct it from the fund
        currentStake = math.min(stake, currentFund)
        currentFund -= currentStake

        // Play the game
        val (winnings) = playGame(currentStake)

        // Add any winnings
        currentFund += winnings

        // Increment the loop counter
        currentGame += 1
    }
    (currentFund)
}

STARTING_FUND: Int = 10
STAKE: Int = 1
NUMBER_OF_GAMES: Int = 25
rollDie: Int
playGame: (stake: Int)Int
playSession: (startingFund: Int, stake: Int, numberOfGames: Int)Int

Enter the following code to play the game 25 times, which is the default value for NUMBER_OF_GAMES.

playSession()

res31: Int = 5

Your bankroll started with a value of 10 units. Is it higher or lower, now?

Now simulate 10,000 players betting 100 chips per game. Play 10,000 games in a session. This Monte Carlo simulation calculates the probability of losing all your money before the end of the session. Enter the follow code:

(sc.parallelize(1 to 10000, 500)
  .map(i => playSession(100000, 100, 250000))
  .map(i => if (i == 0) 1 else 0)
  .reduce(_+_)/10000.0)

res32: Double = 0.9992

Note that the syntax .reduce(_+_) is shorthand in Scala for aggregating by using a summing function.

The preceding code performs the following steps:

1. Creates an RDD with the results of playing the session.

• Replaces bankrupt players' results with the number 1 and nonzero results with the number 0.
• Sums the count of bankrupt players.
• Divides the count by the number of players.

A typical result might be:

res32: Double = 0.9992

Which represents a near guarantee of losing all your money, even though the casino had only a one-percent advantage.

Project Ideas

Try to create a scalable simulation of interest to you.

Here are some mature projects:

• https://github.com/zishanfu/GeoSparkSim
• https://github.com/srbaird/mc-var-spark

See a more complete VaR modeling: - https://databricks.com/blog/2020/05/27/modernizing-risk-management-part-1-streaming-data-ingestion-rapid-model-development-and-monte-carlo-simulations-at-scale.html

ScaDaMaLe Course site and book

Introduction to Machine Learning

Some very useful resources we will weave around for Statistical Learning, Data Mining, Machine Learning:

• January 2014, Stanford University professors Trevor Hastie and Rob Tibshirani (authors of the legendary Elements of Statistical Learning textbook) taught an online course based on their newest textbook, An Introduction to Statistical Learning with Applications in R (ISLR).
• https://www.dataschool.io/15-hours-of-expert-machine-learning-videos/
• free PDF of the ISLR book: http://www-bcf.usc.edu/~gareth/ISL/
• A more theoretically sound book with interesting aplications is Elements of Statistical Learning by the Stanford Gang of 3 (Hastie, Tibshirani and Friedman):
  • free PDF of the 10th printing: http://statweb.stanford.edu/~tibs/ElemStatLearn/printings/ESLII_print10.pdf
  • Solutions: http://waxworksmath.com/Authors/GM/Hastie/WriteUp/weatherwaxepsteinhastiesolutions_manual.pdf
• A great series on Probabilistic ML by Kevin P. Murphy https://probml.github.io/pml-book/.

Deep Learning is a very popular method currently (2017) in Machine Learning and I recommend Andrew Ng's free course in Courseera for this if you have not taken a course in deep learning already.

Note: This is an applied course in data science and we will quickly move to doing things with data. You have to do work to get a deeper mathematical understanding or take other courses. We will focus on intution here and the distributed ML Pipeline in action.

I am assuming several of you have or are taking ML courses. In this course we will focus on getting our hads dirty quickly with some level of common understanding across all the disciplines represented here. Please dive deep at your own time to desired depths and tangents based on your background.

Summary of Machine Learning at a High Level

• A rough definition of machine learning.
  • constructing and studying algorithms that learn from and make predictions on data.
• This broad area involves tools and ideas from various domains, including:
  • computer science,
  • probability and statistics,
  • optimization,
  • linear algebra
  • logic
  • etc.
• Common examples of ML, include:
  • facial recognition,
  • link prediction,
  • text or document classification, eg.::
    • spam detection,
  • protein structure prediction
  • teaching computers to play games (go!)

Some common terminology

using example of spam detection as a running example.

• the data points we learn from are call observations:

  • they are items or entities used for::
    • learning or
    • evaluation.

• in the context of spam detection,

  • emails are our observations.
  • Features are attributes used to represent an observation.
  • Features are typically numeric,
    • and in spam detection, they can be:
    • the length,
    • the date, or
    • the presence or absence of keywords in emails.
  • Labels are values or categories assigned to observations.
    • and in spam detection, they can be:
    • an email being defined as spam or not spam.

• Training and test data sets are the observations that we use to train and evaluate a learning algorithm.

• Pop-Quiz

  • What is the difference between supervised and unsupervised learning?

For a Stats@Stanford Hastie-Tibshirani Perspective on Supervised and Unsupervised Learning:

(watch later 12:12):

From : https://www.dataschool.io/15-hours-of-expert-machine-learning-videos/.

ML Pipelines

Here we will use ML Pipelines to do machine learning at scale.

See https://spark.apache.org/docs/latest/ml-pipeline.html for a quick overview.

Read this section for an overview:

• https://learning.oreilly.com/library/view/spark-the-definitive/9781491912201/ch24.html#high-level-mllib-concepts.

ScaDaMaLe Course site and book

Million Song Dataset - Kaggle Challenge

Predict which songs a user will listen to.

SOURCE: This is just a Scala-rification of the Python notebook published in databricks community edition in 2016.

When you first hear a song, do you ever categorize it as slow or fast in your head? Is it even a valid categorization? If so, can one do it automatically? I have always wondered about that. That is why I got excited when I learned about the Million Songs Dataset -Kaggle Challenge.

In this tutorial we will walk through a practical example of a data science project with Apache Spark in Python. We are going to parse, explore and model a sample from the million songs dataset stored on distributed storage. This tutorial is organized into three sections:

1. ETL: Parses raw texts and creates a cached table
2. Explore: Explores different aspects of the songs table using graphs
3. Model: Uses SparkML to cluster songs based on some of their attributes

The goal of this tutorial is to prepare you for real world data science projects. Make sure you go through the tutorial in the above order and use the exercises to make yourself familiar further with the API. Also make sure you run these notebooks on a 1.6.x cluster.

1. ETL

The first step of most data science projects is extracting, transforming and loading data into well formated tables. Our example starts with ETL as well. By following the ETL noteboook you can expect to learn about following Spark concepts:

• RDD: Resilient Distributed Dataset
• Reading and transforming RDDs
• Schema in Spark
• Spark DataFrame
• Temp tables
• Caching tables

2. Explore

Exploratory analysis is a key step in any real data project. Data scientists use variety of tools to explore and visualize their data. In the second notebook of this tutorial we introduce several tools in Python and Databricks notebooks that can help you visually explore your large data. By reading this notebook and finishing its exercises you will become familiar with:

• How to view the schema of a table
• How to display ggplot and matplotlib figures in Notebooks
• How to summarize and visualize different aspects of large datasets
• How to sample and visualize large data

3. Model

The end goal of many data scientists is producing useful models. These models are often used for prediction of new and upcoming events in production. In our third notebook we construct a simple K-means clustering model. In this notebook you will learn about:

• Feature transformation
• Fitting a model using SparkML API
• Applying a model to data
• Visualizing model results
• Model tuning

ScaDaMaLe Course site and book

Million Song Dataset - Kaggle Challenge

Predict which songs a user will listen to.

SOURCE: This is just a Scala-rification of the Python notebook published in databricks community edition in 2016.

CAUTION: This notebook is expected to have an error in command 28 (Cmd 28 in databricks notebook). You are meant to learn how to fix this error with simple exception-handling to become a better data scientist. So ignore this warning, if any.

Stage 1: Parsing songs data

This is the first notebook in this tutorial. In this notebook we will read data from DBFS (DataBricks FileSystem). We will parse data and load it as a table that can be readily used in following notebooks.

By going through this notebook you can expect to learn how to read distributed data as an RDD, how to transform RDDs, and how to construct a Spark DataFrame from an RDD and register it as a table.

We first explore different files in our distributed file system. We use a header file to construct a Spark Schema object. We write a function that takes the header and casts strings in each line of our data to corresponding types. Once we run this function on the data we find that it fails on some corner caes. We update our function and finally get a parsed RDD. We combine that RDD and the Schema to construct a DataFame and register it as a temporary table in SparkSQL.

Text data files are stored in dbfs:/databricks-datasets/songs/data-001

You can conveniently list files on distributed file system (DBFS, S3 or HDFS) using %fs commands.

ls /databricks-datasets/songs/data-001/

As you can see in the listing we have data files and a single header file. The header file seems interesting and worth a first inspection at first. The file is 377 bytes, therefore it is safe to collect the entire content of the file in the notebook.

sc.textFile("databricks-datasets/songs/data-001/header.txt").collect() 

res1: Array[String] = Array(artist_id:string, artist_latitude:double, artist_longitude:double, artist_location:string, artist_name:string, duration:double, end_of_fade_in:double, key:int, key_confidence:double, loudness:double, release:string, song_hotnes:double, song_id:string, start_of_fade_out:double, tempo:double, time_signature:double, time_signature_confidence:double, title:string, year:double, partial_sequence:int)

Remember you can collect() a huge RDD and crash the driver program - so it is a good practise to take a couple lines and count the number of lines, especially if you have no idea what file you are trying to read.

sc.textFile("databricks-datasets/songs/data-001/header.txt").take(2)

res2: Array[String] = Array(artist_id:string, artist_latitude:double)

sc.textFile("databricks-datasets/songs/data-001/header.txt").count()

res3: Long = 20

sc.textFile("databricks-datasets/songs/data-001/header.txt").collect.map(println) // uncomment to see line-by-line

artist_id:string
artist_latitude:double
artist_longitude:double
artist_location:string
artist_name:string
duration:double
end_of_fade_in:double
key:int
key_confidence:double
loudness:double
release:string
song_hotnes:double
song_id:string
start_of_fade_out:double
tempo:double
time_signature:double
time_signature_confidence:double
title:string
year:double
partial_sequence:int
res4: Array[Unit] = Array((), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), (), ())

As seen above each line in the header consists of a name and a type separated by colon. We will need to parse the header file as follows:

val header = sc.textFile("/databricks-datasets/songs/data-001/header.txt").map(
      line => {
                val headerElement = line.split(":")
                (headerElement(0), headerElement(1))
            }
           ).collect()

header: Array[(String, String)] = Array((artist_id,string), (artist_latitude,double), (artist_longitude,double), (artist_location,string), (artist_name,string), (duration,double), (end_of_fade_in,double), (key,int), (key_confidence,double), (loudness,double), (release,string), (song_hotnes,double), (song_id,string), (start_of_fade_out,double), (tempo,double), (time_signature,double), (time_signature_confidence,double), (title,string), (year,double), (partial_sequence,int))

Let's define a case class called Song that will be used to represent each row of data in the files:

• /databricks-datasets/songs/data-001/part-00000 through /databricks-datasets/songs/data-001/part-00119 or the last .../part-***** file.

case class Song(artist_id: String, artist_latitude: Double, artist_longitude: Double, artist_location: String, artist_name: String, duration: Double, end_of_fade_in: Double, key: Int, key_confidence: Double, loudness: Double, release: String, song_hotness: Double, song_id: String, start_of_fade_out: Double, tempo: Double, time_signature: Double, time_signature_confidence: Double, title: String, year: Double, partial_sequence: Int)

defined class Song

Now we turn to data files. First, step is inspecting the first line of data to inspect its format.

// this is loads all the data - a subset of the 1M songs dataset
val dataRDD = sc.textFile("/databricks-datasets/songs/data-001/part-*") 

dataRDD: org.apache.spark.rdd.RDD[String] = /databricks-datasets/songs/data-001/part-* MapPartitionsRDD[54] at textFile at command-2971213210276292:2

dataRDD.count // number of songs

res5: Long = 31369

dataRDD.take(3)

res6: Array[String] = Array(AR81V6H1187FB48872	nan	nan		Earl Sixteen	213.7073	0.0	11	0.419	-12.106	Soldier of Jah Army	nan	SOVNZSZ12AB018A9B8	208.289	125.882	1	0.0	Rastaman	2003	--, ARVVZQP11E2835DBCB	nan	nan		Wavves	133.25016	0.0	0	0.282	0.596	Wavvves	0.471578247701	SOJTQHQ12A8C143C5F	128.116	89.519	1	0.0	I Want To See You (And Go To The Movies)	2009	--, ARFG9M11187FB3BBCB	nan	nan	Nashua USA	C-Side	247.32689	0.0	9	0.612	-4.896	Santa Festival Compilation 2008 vol.1	nan	SOAJSQL12AB0180501	242.196	171.278	5	1.0	Loose on the Dancefloor	0	225261)

Each line of data consists of multiple fields separated by \t. With that information and what we learned from the header file, we set out to parse our data.

• We have already created a case class based on the header (which seems to agree with the 3 lines above).
• Next, we will create a function that takes each line as input and returns the case class as output.

// let's do this 'by hand' to re-flex our RDD-muscles :)
// although this is not a robust way to read from a data engineering perspective (without fielding exceptions)
def parseLine(line: String): Song = {
  
  val tokens = line.split("\t")
  Song(tokens(0), tokens(1).toDouble, tokens(2).toDouble, tokens(3), tokens(4), tokens(5).toDouble, tokens(6).toDouble, tokens(7).toInt, tokens(8).toDouble, tokens(9).toDouble, tokens(10), tokens(11).toDouble, tokens(12), tokens(13).toDouble, tokens(14).toDouble, tokens(15).toDouble, tokens(16).toDouble, tokens(17), tokens(18).toDouble, tokens(19).toInt)
}

parseLine: (line: String)Song

With this function we can transform the dataRDD to another RDD that consists of Song case classes

val parsedRDD = dataRDD.map(parseLine)

parsedRDD: org.apache.spark.rdd.RDD[Song] = MapPartitionsRDD[55] at map at command-2971213210276298:1

To convert an RDD of case classes to a DataFrame, we just need to call the toDF method

val df = parsedRDD.toDF

df: org.apache.spark.sql.DataFrame = [artist_id: string, artist_latitude: double ... 18 more fields]

Once we get a DataFrame we can register it as a temporary table. That will allow us to use its name in SQL queries.

df.createOrReplaceTempView("songsTable")

We can now cache our table. So far all operations have been lazy. This is the first time Spark will attempt to actually read all our data and apply the transformations.

If you are running Spark 1.6+ the next command will throw a parsing error.

cache table songsTable

The error means that we are trying to convert a missing value to a Double. Here is an updated version of the parseLine function to deal with missing values

// good data engineering science practise
def parseLine(line: String): Song = {
  
  
  def toDouble(value: String, defaultVal: Double): Double = {
    try {
       value.toDouble
    } catch {
      case e: Exception => defaultVal
    }
  }

  def toInt(value: String, defaultVal: Int): Int = {
    try {
       value.toInt
      } catch {
      case e: Exception => defaultVal
    }
  }
  
  val tokens = line.split("\t")
  Song(tokens(0), toDouble(tokens(1), 0.0), toDouble(tokens(2), 0.0), tokens(3), tokens(4), toDouble(tokens(5), 0.0), toDouble(tokens(6), 0.0), toInt(tokens(7), -1), toDouble(tokens(8), 0.0), toDouble(tokens(9), 0.0), tokens(10), toDouble(tokens(11), 0.0), tokens(12), toDouble(tokens(13), 0.0), toDouble(tokens(14), 0.0), toDouble(tokens(15), 0.0), toDouble(tokens(16), 0.0), tokens(17), toDouble(tokens(18), 0.0), toInt(tokens(19), -1))
}

parseLine: (line: String)Song

val df = dataRDD.map(parseLine).toDF
df.createOrReplaceTempView("songsTable")

df: org.apache.spark.sql.DataFrame = [artist_id: string, artist_latitude: double ... 18 more fields]

And let's try caching the table. We are going to access this data multiple times in following notebooks, therefore it is a good idea to cache it in memory for faster subsequent access.

cache table songsTable

From now on we can easily query our data using the temporary table we just created and cached in memory. Since it is registered as a table we can conveniently use SQL as well as Spark API to access it.

select * from songsTable limit 10

Next up is exploring this data. Click on the Exploration notebook to continue the tutorial.

ScaDaMaLe Course site and book

Million Song Dataset - Kaggle Challenge

Predict which songs a user will listen to.

SOURCE: This is just a Scala-rification of the Python notebook published in databricks community edition in 2016.

Stage 2: Exploring songs data

This is the second notebook in this tutorial. In this notebook we do what any data scientist does with their data right after parsing it: exploring and understanding different aspect of data. Make sure you understand how we get the songsTable by reading and running the ETL notebook. In the ETL notebook we created and cached a temporary table named songsTable

Let's Do all the main bits in Stage 1 now before doing Stage 2 in this Notebook.

// Let's quickly do everything to register the tempView of the table here

// fill in comment ... EXERCISE!
case class Song(artist_id: String, artist_latitude: Double, artist_longitude: Double, artist_location: String, artist_name: String, duration: Double, end_of_fade_in: Double, key: Int, key_confidence: Double, loudness: Double, release: String, song_hotness: Double, song_id: String, start_of_fade_out: Double, tempo: Double, time_signature: Double, time_signature_confidence: Double, title: String, year: Double, partial_sequence: Int)

def parseLine(line: String): Song = {
  // fill in comment ...
  
  def toDouble(value: String, defaultVal: Double): Double = {
    try {
       value.toDouble
    } catch {
      case e: Exception => defaultVal
    }
  }

  def toInt(value: String, defaultVal: Int): Int = {
    try {
       value.toInt
      } catch {
      case e: Exception => defaultVal
    }
  }
  // fill in comment ...
  val tokens = line.split("\t")
  Song(tokens(0), toDouble(tokens(1), 0.0), toDouble(tokens(2), 0.0), tokens(3), tokens(4), toDouble(tokens(5), 0.0), toDouble(tokens(6), 0.0), toInt(tokens(7), -1), toDouble(tokens(8), 0.0), toDouble(tokens(9), 0.0), tokens(10), toDouble(tokens(11), 0.0), tokens(12), toDouble(tokens(13), 0.0), toDouble(tokens(14), 0.0), toDouble(tokens(15), 0.0), toDouble(tokens(16), 0.0), tokens(17), toDouble(tokens(18), 0.0), toInt(tokens(19), -1))
}

// this is loads all the data - a subset of the 1M songs dataset
val dataRDD = sc.textFile("/databricks-datasets/songs/data-001/part-*") 

// .. fill in comment
val df = dataRDD.map(parseLine).toDF

// .. fill in comment
df.createOrReplaceTempView("songsTable")

defined class Song
parseLine: (line: String)Song
dataRDD: org.apache.spark.rdd.RDD[String] = /databricks-datasets/songs/data-001/part-* MapPartitionsRDD[81] at textFile at command-2971213210276755:30
df: org.apache.spark.sql.DataFrame = [artist_id: string, artist_latitude: double ... 18 more fields]

spark.catalog.listTables.show(false) // make sure the temp view of our table is there

+----------------------------+--------+-----------+---------+-----------+
|name                        |database|description|tableType|isTemporary|
+----------------------------+--------+-----------+---------+-----------+
|countrycodes                |default |null       |EXTERNAL |false      |
|ltcar_locations_2_csv       |default |null       |MANAGED  |false      |
|magellan                    |default |null       |MANAGED  |false      |
|mobile_sample               |default |null       |EXTERNAL |false      |
|over300all_2_txt            |default |null       |MANAGED  |false      |
|person                      |default |null       |MANAGED  |false      |
|persons                     |default |null       |MANAGED  |false      |
|social_media_usage          |default |null       |MANAGED  |false      |
|social_media_usage_csv_gui  |default |null       |MANAGED  |false      |
|voronoi20191213uppsla1st_txt|default |null       |MANAGED  |false      |
|voronoi20191213uppsla2d_txt |default |null       |MANAGED  |false      |
|voronoi20191213uppsla3d_txt |default |null       |MANAGED  |false      |
|songstable                  |null    |null       |TEMPORARY|true       |
+----------------------------+--------+-----------+---------+-----------+

A first inspection

A first step to any data exploration is viewing sample data. For this purpose we can use a simple SQL query that returns first 10 rows.

select * from songsTable limit 10

table("songsTable").printSchema()

root
 |-- artist_id: string (nullable = true)
 |-- artist_latitude: double (nullable = false)
 |-- artist_longitude: double (nullable = false)
 |-- artist_location: string (nullable = true)
 |-- artist_name: string (nullable = true)
 |-- duration: double (nullable = false)
 |-- end_of_fade_in: double (nullable = false)
 |-- key: integer (nullable = false)
 |-- key_confidence: double (nullable = false)
 |-- loudness: double (nullable = false)
 |-- release: string (nullable = true)
 |-- song_hotness: double (nullable = false)
 |-- song_id: string (nullable = true)
 |-- start_of_fade_out: double (nullable = false)
 |-- tempo: double (nullable = false)
 |-- time_signature: double (nullable = false)
 |-- time_signature_confidence: double (nullable = false)
 |-- title: string (nullable = true)
 |-- year: double (nullable = false)
 |-- partial_sequence: integer (nullable = false)

select count(*) from songsTable

table("songsTable").count() // or equivalently with DataFrame API - recall table("songsTable") is a DataFrame

res4: Long = 31369

display(sqlContext.sql("SELECT duration, year FROM songsTable")) // Aggregation is set to 'Average' in 'Plot Options'